package async

import (
	"context"
	_ "embed"
	"github.com/redis/go-redis/v9"
	"strconv"
	"time"
	"webook/internal/domain"
	"webook/internal/repository"
	"webook/internal/service/sms"
	"webook/pkg/limiter"
	"webook/pkg/logger"
)

//go:embed lua/service_crash.lua
var serviceCrashLuaScript string

type Service struct {
	svc sms.Service
	// 转异步，存储发短信请求的 repository
	repo       repository.AsyncSmsRepository
	l          logger.LoggerV1
	cmd        redis.Cmdable
	limiter    limiter.Limiter
	limiterKey string //限流的key

	serviceCrashStatusRedisKey string        //判断服务是否崩溃的redis key
	errorRateThreshold         int           //错误率阈值（百分比）
	absRTThresholdMs           int           //绝对响应时间阈值（毫秒）
	serviceCrashWindowsTime    time.Duration //时间窗口大小（秒）

	maximumAsyncHoldTime time.Duration //进入异步状态多少分钟后就退出
}

func NewService(svc sms.Service, repo repository.AsyncSmsRepository, l logger.LoggerV1,
	limiter limiter.Limiter, cmd redis.Cmdable, serviceCrashStatusRedisKey string,
	errorRateThreshold, absRTThresholdMs int, serviceCrashWindowsTime time.Duration) *Service {
	res := &Service{svc: svc, repo: repo, l: l,
		limiter: limiter, limiterKey: "sms-limiter",
		cmd:                        cmd,
		serviceCrashStatusRedisKey: "sms-resp-status",
		errorRateThreshold:         errorRateThreshold,
		absRTThresholdMs:           absRTThresholdMs,
		serviceCrashWindowsTime:    serviceCrashWindowsTime,
	}
	go func() {
		res.StartAsyncCycle()
	}()
	return res
}

// 异步发送消息 开一个协程执行此操作 抢占式调度
func (s *Service) StartAsyncCycle() {
	// 这个是我为了测试而引入的，防止你在运行测试的时候，会出现偶发性的失败
	time.Sleep(time.Second * 3)
	for { //不断地循环
		s.AsyncSend()
	}
}

func (s *Service) AsyncSend() {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	// 抢占一个异步发送的消息，确保在非常多个实例
	// 比如 k8s 部署了三个 pod，一个请求，只有一个实例能拿到
	as, err := s.repo.PreemptWaitingSMS(ctx)
	cancel()
	switch err {
	case nil: //拿到了发送短信的任务
		startTime := time.Now()
		ctx, cancel = context.WithTimeout(context.Background(), time.Second)
		defer cancel()
		err = s.svc.Send(ctx, as.TplId, as.Args, as.Numbers...)
		if err != nil {
			// 啥也不需要干
			s.l.Error("执行异步发送短信失败", logger.Error(err), logger.Int64("id", as.Id))
		}
		s.saveSendStatus2Redis(ctx, time.Since(startTime).Milliseconds(), err == nil)
		res := err == nil
		// 通知 repository 我这一次的执行结果
		err = s.repo.ReportScheduleResult(ctx, as.Id, res)
		if err != nil {
			s.l.Error("执行异步发送短信成功，但是标记数据库失败", logger.Error(err), logger.Bool("res", res), logger.Int64("id", as.Id))
		}
	case repository.ErrWaitingSMSNotFound:
		time.Sleep(time.Second)
	default:
		// 正常来说应该是数据库那边出了问题，
		// 睡眠的话可以规避掉短时间的网络抖动问题
		s.l.Error("抢占异步发送短信任务失败", logger.Error(err))
		time.Sleep(time.Second)
	}
}

func (s *Service) Send(ctx context.Context, tplId string, args []string, numbers ...string) error {
	if s.needAsync(ctx) {
		// 需要异步发送，直接转储到数据库
		err := s.repo.Add(ctx, domain.AsyncSms{
			TplId:   tplId,
			Args:    args,
			Numbers: numbers,
			// 设置可以重试三次
			RetryMax: 3,
		})
		return err
	}
	startTime := time.Now()
	err := s.svc.Send(ctx, tplId, args, numbers...)
	go s.saveSendStatus2Redis(ctx, time.Since(startTime).Milliseconds(), err == nil)
	return err
}

// 开始思考系统容错问题 面试装逼，赢得竞争优势就靠这个
func (s *Service) needAsync(ctx context.Context) bool {
	// 这边就是你要设计的，各种判定要不要触发异步的方案
	// 1. 基于响应时间的，平均响应时间:比如当前秒比上一秒平均rt超过30% 或是高于1s
	// 1.1 使用绝对阈值，比如说直接发送的时候，（连续一段时间，或者连续N个请求）响应时间超过了 500ms，然后后续请求转异步
	// 1.2 变化趋势，比如说当前一秒钟内的所有请求的响应时间比上一秒钟增长了 X%，就转异步
	// 2. 基于错误率：一段时间内，收到 err 的请求比率大于 X%，转异步
	// 什么时候退出异步
	// 1. 进入异步 N 分钟后
	// 2. 保留 1% 的流量（或者更少），继续同步发送，判定响应时间/错误率

	//1.触发了限流
	limited, err := s.limiter.Limit(ctx, s.limiterKey)
	if err != nil { //查询限流状态失败 为了稳妥返回限流
		s.l.Error("查询限流状态失败", logger.Error(err), logger.String("limiterKey", s.limiterKey))
		return true
	}
	if limited { //触发限流
		return true
	}
	//2.判定服务商崩溃
	if s.checkServiceIsCrashed(ctx) {
		return true
	}
	return false
}
func (s *Service) checkServiceIsCrashed(ctx context.Context) bool {
	isCrashed, err := s.cmd.Eval(ctx, serviceCrashLuaScript, []string{s.serviceCrashStatusRedisKey},
		s.errorRateThreshold, s.absRTThresholdMs, s.serviceCrashWindowsTime.Milliseconds(),
		time.Now().UnixMilli()).Bool()
	if err != nil {
		s.l.Error("判断崩溃失败", logger.Error(err))
		isCrashed = true //设置为崩溃
	}
	return isCrashed
}
func (s *Service) saveSendStatus2Redis(ctx context.Context, costTimeMs int64, status bool) {
	now := time.Now()
	var tmp string
	if status {
		tmp = strconv.FormatInt(costTimeMs, 10) + ":success"
	} else {
		tmp = strconv.FormatInt(costTimeMs, 10) + ":error"
	}

	err := s.cmd.ZAdd(ctx, s.serviceCrashStatusRedisKey, redis.Z{
		Score:  float64(now.UnixMilli()),
		Member: tmp,
	}).Err()
	if err != nil {
		s.l.Error("保存同步状态到redis失败", logger.Error(err))
	}
}
